package edu.utexas.mapreduce;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

public class InputSplitterFixedSize implements InputSplitter {

	final int blockSize;
	InputSplitterFixedSize(int blockSize) {
		this.blockSize = blockSize;
	}
	
	public List<InputBlock> splitInput(File inputFile) {
		
		String fileName = inputFile.getAbsolutePath();
		long fileLength = inputFile.length();
		long blockLength = this.blockSize;
		List<InputBlock> blockList = new ArrayList<InputBlock>();
		long splitCount = (fileLength + blockLength - 1) / blockLength;
		
		for (int i = 0; i < splitCount; i++) {
			long start = i * blockLength;
			long end = Math.min((i + 1) * blockLength, fileLength);	
			blockList.add(new InputBlock(fileName, start, end));
		}
		
		// if there's a single split, we are done. 
		if (splitCount == 1) {
			return blockList;
		}
		
		// check if the last block is 'small' (half the block size)
		// if so, merge the last two blocks
		int endIndex = blockList.size() - 1;
		InputBlock lastBlock = blockList.get(endIndex);
		boolean mergeLast = (lastBlock.endOffset - lastBlock.startOffset) < (blockLength / 2);
		if (mergeLast) {
			InputBlock endBlock = blockList.remove(endIndex);
			InputBlock startBlock = blockList.remove(endIndex - 1);
			blockList.add(new InputBlock(fileName, startBlock.startOffset, endBlock.endOffset));
		}
		return blockList;
	}
}
